package cdrprocessing.bolt.stat;

import java.util.Map;

import backtype.storm.Config;
import backtype.storm.Constants;
import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.topology.base.BaseBasicBolt;
import backtype.storm.tuple.Tuple;

public abstract class AbstractTimeWindowBolt extends BaseBasicBolt {

    public Map<String, Object> getComponentConfiguration() {
        final Config conf = new Config();
        conf.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS, getTimeWindowSize());
        return conf;
    }

    public void execute(Tuple input, BasicOutputCollector collector) {
        if (isTickTuple(input)) {
            handleTickTuple(input, collector);
        } else {
            handleNormalTuple(input, collector);
        }
    }

    protected abstract void handleNormalTuple(Tuple input, BasicOutputCollector collector);

    protected abstract void handleTickTuple(Tuple input, BasicOutputCollector collector);

    protected abstract int getTimeWindowSize();

    private boolean isTickTuple(Tuple tuple) {
        return tuple.getSourceComponent().equals(Constants.SYSTEM_COMPONENT_ID) && tuple.getSourceStreamId().equals(Constants.SYSTEM_TICK_STREAM_ID);
    }
}
